package com.xiaohongshu.kafkademo.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;

import com.xiaohongshu.kafkademo.event.DataChangeEvent;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;


import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

@Component
@Slf4j
public class CanalClient {

    @Resource
    private ApplicationEventPublisher eventPublisher;

    @Value("${canal.server.host:127.0.0.1}")
    private String canalHost;

    @Value("${canal.server.port}")
    private Integer canalPort;

    @Value("${canal.server.destination:example}")
    private String destination;

    @Value("${canal.server.username}")
    private String username;

    @Value("${canal.server.password}")
    private String password;


    @Value("${kafka.topic.my-topic3}")
    String myTopic3;

    @Resource
    private  KafkaTemplate<String, Object> kafkaTemplate;

    private CanalConnector connector;
    private volatile boolean running = false;





    @PostConstruct
    public void init() {
        // 创建Canal连接
        connector = CanalConnectors.newSingleConnector(
            new InetSocketAddress(canalHost, canalPort),
            destination,
            username,
            password
        );
        
        startListener();
    }

    private void startListener() {
        CompletableFuture.runAsync(() -> {
            try {
                connector.connect();
                connector.subscribe("test\\..*"); // 监听test数据库的所有表
                connector.rollback();
                running = true;

                log.info("Canal客户端启动成功，开始监听数据库变更...");
                log.info("监听地址: {}:{}, destination: {}", canalHost, canalPort, destination);

                while (running) {
                    Message message = connector.getWithoutAck(1000);
                    long batchId = message.getId();
                    int size = message.getEntries().size();

                    if (batchId == -1 || size == 0) {
                        Thread.sleep(1000);
                    } else {
                        log.debug("接收到 {} 条数据变更记录", size);
                        handleMessage(message.getEntries());
                    }

                    connector.ack(batchId);
                }
            } catch (Exception e) {
                log.error("Canal监听异常", e);
                // 重连逻辑
                reconnect();
            }
        });
    }

    private void reconnect() {
        try {
            Thread.sleep(5000); // 等待5秒后重连
            if (connector != null) {
                connector.disconnect();
            }
            log.info("尝试重新连接Canal...");
            startListener();
        } catch (Exception e) {
            log.error("重连失败", e);
        }
    }

    private void handleMessage(List<Entry> entries) {
        for (Entry entry : entries) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN ||
                entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            RowChange rowChange;
            try {
                rowChange = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                log.error("解析RowChange失败", e);
                continue;
            }

            EventType eventType = rowChange.getEventType();
            String database = entry.getHeader().getSchemaName();
            String table = entry.getHeader().getTableName();
            long executeTime = entry.getHeader().getExecuteTime();

            log.info("检测到数据变更 - 数据库: {}, 表: {}, 事件类型: {}, 执行时间: {}", 
                     database, table, eventType, executeTime);

            for (RowData rowData : rowChange.getRowDatasList()) {
                processRowData(database, table, eventType.name(), rowData, executeTime);
            }
        }
    }

    private void processRowData(String database, String table, String eventType, 
                               RowData rowData, long executeTime) {
        Map<String, Object> beforeData = new HashMap<>();
        Map<String, Object> afterData = new HashMap<>();
        Long id = null;

        // 处理变更前数据
        if (rowData.getBeforeColumnsList() != null && !rowData.getBeforeColumnsList().isEmpty()) {
            for (Column column : rowData.getBeforeColumnsList()) {
                beforeData.put(column.getName(), column.getValue());
                if ("id".equals(column.getName())) {
                    id = Long.valueOf(column.getValue());
                }
            }
        }

        // 处理变更后数据
        if (rowData.getAfterColumnsList() != null && !rowData.getAfterColumnsList().isEmpty()) {
            for (Column column : rowData.getAfterColumnsList()) {
                afterData.put(column.getName(), column.getValue());
                if ("id".equals(column.getName())) {
                    id = Long.valueOf(column.getValue());
                }
            }
        }

        // 创建变更消息对象
        Map<String, Object> changeMessage = new HashMap<>();
        changeMessage.put("database", database);
        changeMessage.put("table", table);
        changeMessage.put("eventType", eventType);
        changeMessage.put("beforeData", beforeData);
        changeMessage.put("afterData", afterData);
        changeMessage.put("executeTime", executeTime);
        changeMessage.put("id", id);

        try {

            String key = database + "." + table + "." + id;
            //发送到kafka 对应的topic
            // 分区编号最好为 null，交给 kafka 自己去分配
            ProducerRecord<String, Object> producerRecord = new ProducerRecord<>(myTopic3, null, System.currentTimeMillis(),key
                    , changeMessage);

            CompletableFuture<SendResult<String, Object>> future = kafkaTemplate.send(producerRecord);
            if (future.get().getRecordMetadata() != null) {
                log.info("生产者成功发送消息到" + future.get().getRecordMetadata().topic() + "-> " + future.get().getRecordMetadata().partition());
            }
        } catch (Exception e ) {
            log.error("canal监听binlog的消息发送至kafka失败！");
            e.printStackTrace();
        }

        // 发布数据变更事件
        DataChangeEvent event = new DataChangeEvent(
            database, table, eventType,
            beforeData, afterData, executeTime, id
        );

        log.debug("发布数据变更事件: {}", event);
        eventPublisher.publishEvent(event);
    }

    @PreDestroy
    public void destroy() {
        running = false;
        if (connector != null) {
            connector.disconnect();
        }
        log.info("Canal客户端已关闭");
    }
}
